package com.garden.rocketmq.learn.samples.config.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Garden on 2021-01-23 14:03
 */
@Configuration
public class TransactionProducerConfig {

    public static final Logger LOGGER = LoggerFactory.getLogger(TransactionProducerConfig.class);

    @Value("${rocketmq.producer.transaction.groupName}")
    private String groupName;
    @Value("${rocketmq.producer.transaction.namesrvAdrr}")
    private String namesrvAddr;

    @Bean(name = "transactionProducer")
    @ConditionalOnMissingBean
    public TransactionMQProducer transactionProducer() throws RuntimeException {

        //构建事务生产者对象，设置属性
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer(this.groupName);
        //设置NameServer的地址，Producer启动后通过NameServer获取目标Topic的路由信息
        transactionMQProducer.setNamesrvAddr(this.namesrvAddr);
        //设置线程池，
        //用于处理执行本地事务[服务端broker存储haft消息成功后请求]或者检查本地事务逻辑[服务端broker接收Pending消息后回查请求]提供线程
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        transactionMQProducer.setExecutorService(executorService);
        //设置本地事务监听器，用于处理执行本地事务或者检查本地事务逻辑
        transactionMQProducer.setTransactionListener(new TransactionListener(){
            private AtomicInteger transactionIndex = new AtomicInteger(0);
            private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                //执行本地事务
                //记录消息事务id以及自定义状态值，返回Pending状态触发回查机制
                int status = (transactionIndex.getAndIncrement()) % 3;
                localTrans.put(message.getTransactionId(),status);
                return LocalTransactionState.UNKNOW;
            }
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                //检查本地事务
                //从本地事务集合中获取目标消息事务对应的自定义状态值，根据自定义状态值返回事务状态
                Integer status = localTrans.get(messageExt.getTransactionId());
                if (null != status) {
                    switch (status) {
                        case 0:
                            return LocalTransactionState.UNKNOW;
                        case 1:
                            return LocalTransactionState.COMMIT_MESSAGE;
                        case 2:
                            return LocalTransactionState.ROLLBACK_MESSAGE;
                        default:
                    }
                }
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        //启动实例
        try {
            transactionMQProducer.start();
            LOGGER.info("transaction producer start success... groupName={},namesrvAdrr={}",this.groupName,this.namesrvAddr);
        } catch (MQClientException e) {
            LOGGER.error("transaction producer start failed... errorMsg={}",e.getErrorMessage());
            throw new RuntimeException(e);
        }

        //返回Bean
        return transactionMQProducer;
    }

}
